package com.gjy.kafka.k10;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.assertj.core.util.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-03 11:15:38
 */
public class KafkaConsumerApp {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerApp.class);
    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final String TOPIC = "test-topic";
    private static final String GROUP_ID = "consumer-group";

    private final KafkaConsumerService consumerService;
    private volatile boolean running = true;
    private final AtomicLong messageCount = new AtomicLong(0);
    private final AtomicLong batchCount = new AtomicLong(0);

    public KafkaConsumerApp() {
        Properties props = KafkaConfig.getManualCommitConsumerConfig(GROUP_ID);
        this.consumerService = new KafkaConsumerService(props);

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }


    /**
     * 启动消费者 - 自动提交模式
     */
    public void startAutoCommit() {
        logger.info("启动自动提交消费者，主题: {}", TOPIC);

        new Thread(() -> {
            consumerService.consumeAutoCommit(TOPIC);
        }, "AutoCommit-Consumer").start();
    }

    /**
     * 启动消费者 - 手动同步提交模式
     */
    public void startManualSyncCommit() {
        logger.info("启动手动同步提交消费者，主题: {}", TOPIC);

        new Thread(() -> {
            try {
                consumerService.consumeManualSyncCommit(TOPIC);
            } catch (Exception e) {
                logger.error("消费者异常: {}", e.getMessage(), e);
            }
        }, "ManualSync-Consumer").start();
    }

    /**
     * 启动消费者 - 手动异步提交模式
     */
    public void startManualAsyncCommit() {
        logger.info("启动手动异步提交消费者，主题: {}", TOPIC);

        new Thread(() -> {
            try {
                consumerService.consumeManualAsyncCommit(TOPIC);
            } catch (Exception e) {
                logger.error("消费者异常: {}", e.getMessage(), e);
            }
        }, "ManualAsync-Consumer").start();
    }

    /**
     * 启动自定义消费者（带消息处理逻辑）
     */
    public void startCustomConsumer() {
        logger.info("启动自定义消费者，主题: {}", TOPIC);

        new Thread(() -> {
            Properties props = KafkaConfig.getManualCommitConsumerConfig(GROUP_ID + "-custom");
            try {
                KafkaConsumerService customConsumer = new KafkaConsumerService(props);
                // 订阅主题
                customConsumer.getConsumer().subscribe(Lists.newArrayList(TOPIC));

                logger.info("自定义消费者开始消费消息...");

                while (running) {
                    try {
                        // 轮询消息
                        ConsumerRecords<String, String> records = customConsumer.getConsumer().poll(java.time.Duration.ofMillis(1000));

                        if (!records.isEmpty()) {
                            logger.info("📥 收到 {} 条消息", records.count());

                            for (ConsumerRecord<String, String> record : records) {
                                processMessageWithStatistics(record);
                            }

                            // 手动提交偏移量
                            customConsumer.getConsumer().commitSync();
                            logger.info("✅ 偏移量提交完成");
                        }

                        // 每处理10条消息打印一次统计
                        if (messageCount.get() % 10 == 0 && messageCount.get() > 0) {
                            printStatistics();
                        }

                    } catch (Exception e) {
                        logger.error("消费消息异常: {}", e.getMessage(), e);
                        // 短暂休眠后继续
                        Thread.sleep(1000);
                    }
                }

            } catch (Exception e) {
                logger.error("自定义消费者异常: {}", e.getMessage(), e);
            }
        }, "Custom-Consumer").start();
    }

    /**
     * 处理消息并统计
     */
    private void processMessageWithStatistics(ConsumerRecord<String, String> record) {
        long count = messageCount.incrementAndGet();

        // 处理消息内容
        String processedMessage = processMessage(record.value());

        logger.info("🎯 消费消息 [#{}] - Partition: {}, Offset: {}, Key: {}, Value: {}, Processed: {}",
                count, record.partition(), record.offset(), record.key(),
                record.value(), processedMessage);

        // 处理headers（如果有）
        if (record.headers() != null) {
            record.headers().forEach(header -> {
                logger.info("   📋 Header - {}: {}", header.key(), new String(header.value()));
            });
        }

        // 模拟处理时间
        try {
            Thread.sleep(100); // 100ms处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 消息处理逻辑（可根据需要自定义）
     */
    private String processMessage(String message) {
        // 这里可以添加各种消息处理逻辑
        // 例如：数据转换、验证、业务处理等

        String processed = message + " [已处理@" + sdf.format(new Date()) + "]";

        // 模拟一些处理逻辑
        if (message.contains("批量")) {
            batchCount.incrementAndGet();
            processed = processed + " [批量消息]";
        }

        if (message.contains("异步")) {
            processed = processed + " [异步消息]";
        }

        return processed;
    }

    /**
     * 打印消费统计
     */
    private void printStatistics() {
        logger.info("📊 消费统计 - 总消息数: {}, 批量消息数: {}, 时间: {}",
                messageCount.get(), batchCount.get(), sdf.format(new Date()));
    }

    /**
     * 从特定分区消费
     */
    public void startPartitionConsumer(List<Integer> partitions) {
        logger.info("启动分区消费者，主题: {}, 分区: {}", TOPIC, partitions);

        new Thread(() -> {
            try {
                consumerService.consumeSpecificPartitions(TOPIC, partitions);
            } catch (Exception e) {
                logger.error("分区消费者异常: {}", e.getMessage(), e);
            }
        }, "Partition-Consumer").start();
    }

    /**
     * 优雅关闭
     */
    public void shutdown() {
        running = false;
        logger.info("正在关闭消费者应用...");

        try {
            // 等待一段时间让消费者完成当前处理
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        consumerService.stop();
        consumerService.close();

        logger.info("消费者应用已关闭");
        logger.info("最终统计 - 总消费消息数: {}", messageCount.get());
    }

    /**
     * 主函数 - 消费者应用入口
     */
    public static void main(String[] args) {
        KafkaConsumerApp app = new KafkaConsumerApp();

        try {
            logger.info("Kafka消费者应用启动，主题: {}", TOPIC);

            // 选择消费模式（取消注释你想要使用的模式）

            // 模式1: 自动提交（简单但可能重复消费）
            // app.startAutoCommit();

            // 模式2: 手动同步提交（推荐用于生产环境）
            // app.startManualSyncCommit();

            // 模式3: 手动异步提交
            // app.startManualAsyncCommit();

            // 模式4: 自定义消费者（推荐，包含完整处理逻辑）
            app.startCustomConsumer();

            // 模式5: 指定分区消费
            // app.startPartitionConsumer(List.of(0, 1));

            // 保持主线程运行
            logger.info("消费者应用运行中... 按 Ctrl+C 停止");

            // 主循环，定期打印统计信息
            while (app.running) {
                try {
                    Thread.sleep(30000); // 每30秒打印一次统计
                    app.printStatistics();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }

        } catch (Exception e) {
            logger.error("消费者应用异常: {}", e.getMessage(), e);
        } finally {
            app.shutdown();
        }
    }

}
